package com.heima.schedule.service.impl;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.heima.common.constants.ScheduleConstants;
import com.heima.common.exception.CustomException;
import com.heima.common.redis.CacheService;
import com.heima.model.common.enums.AppHttpCodeEnum;
import com.heima.model.schedule.dtos.Task;
import com.heima.model.schedule.pojos.Taskinfo;
import com.heima.model.schedule.pojos.TaskinfoLogs;
import com.heima.schedule.mapper.TaskInfoLogsMapper;
import com.heima.schedule.mapper.TaskInfoMapper;
import com.heima.schedule.service.TaskService;
import lombok.extern.slf4j.Slf4j;
import org.joda.time.DateTime;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

import java.util.Date;
import java.util.List;
import java.util.Set;

/**
 * @author brianxia
 * @version 1.0
 * @date 2022-11-25 11:20
 */
@Service
@Slf4j
public class TaskServiceImpl implements TaskService {
    @Autowired
    private TaskInfoLogsMapper taskInfoLogsMapper;
    @Autowired
    private TaskInfoMapper taskInfoMapper;
    @Autowired
    private CacheService cacheService;

    //1.添加任务

    @Override
    @Transactional(rollbackFor = Exception.class)
    public void add(Task task) {
        //1.1 添加任务到shujuku
        addTaskToDB(task);

        //1.2 保存任务到缓存中
        addTaskToCache(task);
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public void cancel(long taskId) {
        //1.取消数据库
        Task task = updateDbStatus(taskId,ScheduleConstants.CANCELLED);

        //2.取消redis
        cancelCache(task);
    }

    @Override
    public Task poll(int priority,int type) {
        String key = ScheduleConstants.TOPIC + priority + "_" + type;
        String taskId = cacheService.lRightPop(key);
        if(StringUtils.isEmpty(taskId)){
            return null;
        }
        String taskStr = (String) cacheService.hGet(ScheduleConstants.HASH_NAME, taskId);
        Task task = JSON.parseObject(taskStr, Task.class);
        //任务删除
        updateDbStatus(Long.parseLong(taskId),ScheduleConstants.EXECUTED);
        //删除缓存中的数据
        cancelCache(task);
        return task;
    }

    private void cancelCache(Task task) {
        //保存缓存
        long executeTime = task.getExecuteTime(); //任务执行时间
        long currentTimeMillis = System.currentTimeMillis(); //当前时间
        long time5Minuts = DateTime.now().plusMinutes(5).toDate().getTime(); //5分钟之后的时间

        if(executeTime <= currentTimeMillis){
            String key = ScheduleConstants.TOPIC + task.getPriority() + "_" + task.getTaskType();
            //立即执行 存放到list
            cacheService.lRemove(key,0, task.getTaskId().toString()); //放入list数据量很大
            cacheService.hDelete(ScheduleConstants.HASH_NAME,task.getTaskId().toString());
        }else if(executeTime <= time5Minuts){
            //5分钟之内数据
            String key = ScheduleConstants.FUTURE + task.getPriority() + "_" + task.getTaskType();
            cacheService.zRemove(key,task.getTaskId().toString());
            cacheService.hDelete(ScheduleConstants.HASH_NAME,task.getTaskId().toString());
        }
    }

    private Task updateDbStatus(long taskId, int status) {
        //1.主表数据直接删除
        taskInfoMapper.deleteById(taskId);
        //2.日志表 修改状态
        TaskinfoLogs taskinfoLogs = taskInfoLogsMapper.selectById(taskId);
        taskinfoLogs.setStatus(status);
        int i = taskInfoLogsMapper.updateById(taskinfoLogs);
        //乐观锁接收返回值
        if(i == 0){
            log.error("当前数据已经被修改" + taskId);
            throw new CustomException(AppHttpCodeEnum.SERVER_ERROR);
        }

        Task task = new Task();
        BeanUtils.copyProperties(taskinfoLogs,task);
        task.setExecuteTime(taskinfoLogs.getExecuteTime().getTime());
        return task;
    }

    private void addTaskToCache(Task task) {
        //保存缓存
        long executeTime = task.getExecuteTime(); //任务执行时间
        long currentTimeMillis = System.currentTimeMillis(); //当前时间
        long time5Minuts = DateTime.now().plusMinutes(5).toDate().getTime(); //5分钟之后的时间

        if(executeTime <= time5Minuts){
            //5分钟之内的数据
            if(executeTime <= currentTimeMillis){
                //当前要执行的数据
                String key = ScheduleConstants.TOPIC + task.getPriority() + "_" + task.getTaskType();
                //立即执行 存放到list
                cacheService.lLeftPush(key, task.getTaskId().toString()); //放入list数据量很大
            }else{
                //5分钟之内，大于当前时间
                String key = ScheduleConstants.FUTURE + task.getPriority() + "_" + task.getTaskType();
                cacheService.zAdd(key,task.getTaskId().toString(),executeTime);
            }
            cacheService.hPut(ScheduleConstants.HASH_NAME,task.getTaskId().toString(),
                    JSON.toJSONString(task));

            //更新到数据库in_redis改成1
            taskInfoMapper.update(null, Wrappers.<Taskinfo>lambdaUpdate()
                    .eq(Taskinfo::getTaskId,task.getTaskId())
                    .set(Taskinfo::getInRedis,1)
            );
        }

    }

    private void addTaskToDB(Task task) {
        //1.任务主表
        Taskinfo taskinfo = new Taskinfo();
        BeanUtils.copyProperties(task,taskinfo);
        taskinfo.setExecuteTime(new Date(task.getExecuteTime()));
        taskInfoMapper.insert(taskinfo);        //拿到雪花算法主键id

        //给task对象赋值id
        task.setTaskId(taskinfo.getTaskId());

        //2.任务日志表
        TaskinfoLogs taskinfoLogs = new TaskinfoLogs();
        BeanUtils.copyProperties(taskinfo,taskinfoLogs);
        taskinfoLogs.setVersion(1); //版本号初始值为1
        taskinfoLogs.setStatus(ScheduleConstants.SCHEDULED); //任务状态初始值为0，待处理
        taskInfoLogsMapper.insert(taskinfoLogs);
    }

    @Scheduled(cron = "0 * * ? * *") //每分钟执行一次，将zset中已经到期数据同步到list
    @Override
    public void syncZsetToList(){

        String token = cacheService.tryLock("myzsettolist109", 30 * 1000L);
        if(!StringUtils.isEmpty(token)){
            System.out.println("同步任务开始执行...");
            //1.获取到所有zset
            Set<String> zsetName = cacheService.scan(ScheduleConstants.FUTURE + "*");

            //2.遍历所有zset将其中数据同步到list
            for (String s : zsetName) {
                //获取0到当前时间zset中的数据，这些数据就是已经到期的数据
                Set<String> taskIds = cacheService.zRangeByScore(s, 0, System.currentTimeMillis());
                if(CollectionUtils.isEmpty(taskIds)){
                    continue;
                }
                //获取list名字 future_50_1 -> topic_50_1
                String topic = s.replace(ScheduleConstants.FUTURE,ScheduleConstants.TOPIC);

                cacheService.refreshWithPipeline(s,topic,taskIds);
//            //添加到list
//            cacheService.lLeftPushAll(topic,taskIds);
//            //删除zset中的数据
//            cacheService.zRemove(s,taskIds);
            }
        }

    }


    @Override
    @Scheduled(cron = "0 * * ? * *") //每分钟执行一次，将zset中已经到期数据同步到list
    public void syncDbToRedis(){

        String token = cacheService.tryLock("syncDbToRedis109", 30 * 1000L);
        if(!StringUtils.isEmpty(token)){
            System.out.println("同步任务开始执行...");

            //1.查询数据库中in_redis为0的数据，加上时间限制
            Date time5Minuts = DateTime.now().plusMinutes(5).toDate(); //5分钟之后的时间
            List<Taskinfo> taskinfos = taskInfoMapper.selectList(Wrappers.<Taskinfo>lambdaQuery()
                    .eq(Taskinfo::getInRedis, 0)
                    .le(Taskinfo::getExecuteTime, time5Minuts)
            );
            if(!CollectionUtils.isEmpty(taskinfos)){
                for (Taskinfo taskinfo : taskinfos) {
                    //添加到缓存中
                    addTaskToCache(taskInfoToTask(taskinfo));
                }
            }
        }

    }

    private Task taskInfoToTask(Taskinfo taskinfo) {
        Task task = new Task();
        BeanUtils.copyProperties(taskinfo,task);
        task.setExecuteTime(taskinfo.getExecuteTime().getTime());
        return task;
    }
}
